/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.sss.client.DataPutter;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.datatype.PackableBytes;

public class PutFileList {

  public static void usage() {
    System.out.println("java PutFileList [-o gid] filelists");
    System.out.println("  each filelist is a text file which contains one filename per line.");
    System.exit(-1);
  }

  public static byte[] readFile(String pathname) throws IOException {
    File file = new File(pathname);
    FileInputStream fin = new FileInputStream(file);
    BufferedInputStream in = new BufferedInputStream(fin);
    byte[] buffer = new byte[(int)file.length()];

    in.read(buffer);
    in.close();
    fin.close();

    return buffer;
  }

  private static class ParsedInfo {
    public List<File> files = new ArrayList<File>();
    public GroupID gid = null;
  }

  private static ParsedInfo parseArgs(String[] args) {
    try {
      ParsedInfo info = new ParsedInfo();
      for (int i=0; i<args.length; i++) {
        if (args[i].equals("-o")) {
          if (i+1 >= args.length) {
            System.err.println("runaway argument.");
            return null;
          }
          try {
            if (info.gid == null) {
              info.gid = GroupID.createFromString(args[i + 1]);
            }
            else {
              System.err.println("duplicate options `-o' are specified.");
              return null;
            }
          }
          catch (IllegalArgumentException e) {
            System.err.println("wrong GroupID specified: " + args[i + 1]);
            return null;
          }
          i++;
        }
        else {
          File list = new File(args[i]);
          int line = 0;
          if (!list.isFile()) {
            System.err.println("file not found: " + args[i]);
            return null;
          }
          else if (!list.canRead()) {
            System.err.println("file cannot be read: " + args[i]);
            return null;
          }
          System.out.println("adding files in list: " + list);
          BufferedReader in;
          in = new BufferedReader(new FileReader(list));
          try {
            String file;
            while ((file = in.readLine()) != null) {
              File f = new File(file);
              if (!f.isFile()) {
                System.err.println("file not found: " + f + " at line: " + line);
                return null;
              }
              else if (!f.canRead()) {
                System.err.println("file cannot be read: " + f + " at line; " + line);
                return null;
              }
              info.files.add(f);
              line++;
            }
          }
          finally {
            in.close();
          }
        }
      }
      return info;
    }
    catch (FileNotFoundException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      return null;
    }
    catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      return null;
    }
  }

  /**
   * Usage:
   * java PutFileList filelist ...
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    ParsedInfo info = parseArgs(others);
    if (info == null || info.files == null || info.files.isEmpty()) {
      usage();
    }

    Map<String, byte[]> cache = new HashMap<String, byte[]>();

    DataPutter<PackableBytes, PackableBytes> putter =
      DataPutter.create(client, PackableBytes.class, PackableBytes.class, info.gid);
    GroupID table = putter.getOutputGroupID();
    try {
      for (File f: info.files) {
        String k = f.getPath();
        byte[] v = cache.get(k);
        if (v == null) {
          v = readFile(k);
          cache.put(k, v);
        }
        putter.append(new PackableBytes(k.getBytes("UTF-8")),
                      new PackableBytes(v));
      }
    }
    finally {
      putter.close();
    }

    System.out.println("output data deployed - " + table.toString());
  }
}
